package com.seaboxdata.jt809.handle;

import com.seaboxdata.jt809.common.util.LonLatUtil;
import com.seaboxdata.jt809.common.util.RedisUtil;
import com.seaboxdata.jt809.kafka.JT809Packet0x1202MessageProducer;
import com.seaboxdata.jt809.packet.JT809Packet0x1202;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;

/**
 * @Author: dfjx
 * @Date: 2020/7/8 16:47
 * @Version 1.0
 * @Describe:车辆定位信息处理类，信息入库
 */
public class JT809Packet0x1202Handle extends SimpleChannelInboundHandler<JT809Packet0x1202> {
    private static Logger log = LoggerFactory.getLogger(JT809Packet0x1202Handle.class);
    private JT809Packet0x1202MessageProducer producer = new JT809Packet0x1202MessageProducer();

    private LonLatUtil.Point[] points = {
            new LonLatUtil.Point(114.502326, 22.671276),
            new LonLatUtil.Point(114.5963682, 22.65788301),
            new LonLatUtil.Point(114.6253057, 22.50513332),
            new LonLatUtil.Point(114.5065689, 22.44014054),
            new LonLatUtil.Point(114.3466708, 22.6012205),
            new LonLatUtil.Point(114.346231, 22.606963),
            new LonLatUtil.Point(114.335217, 22.60711),
            new LonLatUtil.Point(114.335108, 22.628361),
            new LonLatUtil.Point(114.374682, 22.64477251),
            new LonLatUtil.Point(114.384271, 22.64417),
            new LonLatUtil.Point(114.408915, 22.656745),
            new LonLatUtil.Point(114.448514, 22.672634),
            new LonLatUtil.Point(114.4616321, 22.6777162),
            new LonLatUtil.Point(114.502326, 22.671276)};

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, JT809Packet0x1202 msg) {
        //去除不是大鹏新区的数据
        if (LonLatUtil.isPtInPoly(new LonLatUtil.Point(Double.parseDouble(msg.getLon() + "") / 1000000.0, Double.parseDouble(msg.getLat() + "") / 1000000.0), points)) {
            //去掉重复的数据
            msg.setVehicleNo(msg.getVehicleNo().replaceAll("\u0000", ""));
            String value = RedisUtil.getString(msg.getVehicleNo());
            LocalDateTime localSendTime = LocalDateTime.of(msg.getDate(), msg.getTime());
            String sendTimeStr = LocalDateTime.of(msg.getDate(), msg.getTime()).toString();
            if (value == null) {
                producer.sendMessageToKafka(msg);
                RedisUtil.setString(msg.getVehicleNo(), LocalDateTime.of(msg.getDate(), msg.getTime()).toString());
            } else {
                if (!value.equals(sendTimeStr) && localSendTime.compareTo(LocalDateTime.parse(value)) == 1) {
                    producer.sendMessageToKafka(msg);
                    RedisUtil.setString(msg.getVehicleNo(), LocalDateTime.of(msg.getDate(), msg.getTime()).toString());
                }
            }
        }
    }

}
